package zookeeper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import org.apache.thrift.TException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 提供connect接口与zookeeper建立连接
 * 并维护一个zookeeper连接
 * 
 * 使用connect接口与服务器进行链接时会将自己作为watcher，用以watch服务器状态，
 * 同时会开启一个计数模型的线程控制器，控制线程等待服务器状态返回。
 * 当服务器状态返回SyncConnected状态时，代表首次连接成功，这时候计数模型置为0，等待的线程将继续执行，进行注册。
 * 
 * @author Administrator
 *
 */
public class ZK implements Watcher {
    public ZK(String connectStr, int sessionTimeOut) throws IOException, InterruptedException {
        //noinspection NullableProblems
        this(connectStr, null, sessionTimeOut);
    }

    @SuppressWarnings({ "rawtypes" })
	public ZK(String connectStr, Callable callback, int sessionTimeOut) throws IOException, InterruptedException {
        this.connectStr = connectStr;
        this.callback = callback;
        this.sessionTimeOut = sessionTimeOut;
    }
    /**
     * 与zookeeper建立链接，并注册Watch事件应付连接状态的变化，包括节点的变化
     * 
     * 连接后，我们建立了一个计数器模型的线程控制对象CountDownLatch，
     * 用以控制当前线程等待，等待计数被别的线程置为0后，接续执行，做callback调用。
     * 
     * 
     * 
     * @throws Exception
     */
    public void connect() throws Exception {
        log.debug("connecting to {}", connectStr);
        connectedSignal = new CountDownLatch(1);
        this.zooKeeper = new ZooKeeper(connectStr, sessionTimeOut, this);
        connectedSignal.await();
        if (callback != null) callback.call();
    }

    protected void reconnect() throws Exception {
        zooKeeper.close();
        registerNode.clear();
        connect();
    }
    
    /**
     * 观察zookeeper上的某个路径，并取其子节点的data数据存储起来
     * @param path
     * @throws TException
     * @throws InterruptedException
     * @throws KeeperException
     */
    public void observeNonLeafNode(String path, ZKObserveDealer dealer) throws InterruptedException, KeeperException,Exception {
    	Map<String, byte[]> map = getChildrenData( path );
    	if(map == null || map.size() == 0){
    		throw new Exception("no server regist on path "+path);
    	}
    	//先取出所有子节点的名字作为list
    	List<ZKnode> l = new ArrayList<ZKnode>();
    	for(Map.Entry<String, byte[]> entry : map.entrySet() ){
    		l.add( new ZKnode(entry.getKey(), entry.getValue()) );    		
    	}
    	observeServerNonLeafNode.add(path);
    	observeServerNonLeafNodeDealMap.put(path, dealer);
    	observeServerNodeChildrenNameDataListMap.put( path , l);
        
    	dealer.dealChange( l );
    }
    /**
     * 观察zookeeper上的某个路径，取其data数据存储起来
     * @param path
     * @param dealer
     * @throws InterruptedException
     * @throws KeeperException
     */
    public void observeLeafNode(String path, ZKObserveDealer dealer) throws InterruptedException, KeeperException {
    	byte[] b = zooKeeper.getData(path, true, null);
		ZKnode zk = new ZKnode(path, b);
    	observeServerLeafNode.add(path);
    	observeServerLeafNodeDealMap.put(path, dealer);
		observeServerNodeNameDataMap.put(path, zk);
		List<ZKnode> l = new ArrayList<ZKnode>();
		l.add(zk);
		dealer.dealChange(l);
    }
    private void dealNodeChange(String path )throws InterruptedException, KeeperException,Exception {
    	log.debug("dealNodeChange '{}'",  path);
    	if( this.observeServerLeafNode.contains(path) ){
    		observeLeafNode( path, observeServerLeafNodeDealMap.get(path));
    	}
    	//节点发生变化，有责任通知其父path，如果父path有被观察的话
    	int last = path.lastIndexOf("/");
        path = path.substring(0, last);
        dealChildrenNodeChange(path);
    }
    private void dealChildrenNodeChange( String path )throws InterruptedException, KeeperException,Exception {
    	log.debug("dealChildrenNodeChange '{}'",  path);
    	if( this.observeServerNonLeafNode.contains(path) ){
    		observeNonLeafNode(path, this.observeServerNonLeafNodeDealMap.get( path ) );
    	}
    }
    /**
     * 获取某路径下的子节点信息
     * key为子节点名字
     * value为子节点的data，字节数组
     * @param path
     * @return
     * @throws InterruptedException
     * @throws KeeperException
     */
    private Map<String, byte[]> getChildrenData(String path)
            throws InterruptedException, KeeperException {
        Map<String, byte[]> ret = new HashMap<String, byte[]>();
        List<String> nodes = zooKeeper.getChildren(path, true);
        
        if (nodes != null)
            for (String p : nodes){
            	//zooKeeper.getData(path, watcher, stat)
            	byte[] b = zooKeeper.getData(path + "/" + p, true, null);
                ret.put(p , b);
            }
        log.debug("getData({}) ==> {}", path, ret);
        return ret;
    }

    public void close() throws InterruptedException {
        run = false;
        if (zooKeeper != null) {
            zooKeeper.close();
            zooKeeper = null;
        }
    }
    
    /**
     * 本机服务需要注册到zookeeper进行监听
     * @param path
     * @param ip
     * @param port
     * @throws TException
     * @throws InterruptedException
     * @throws KeeperException
     */
    public void register(String path, String ip, String port)
            throws TException, InterruptedException, KeeperException {
        String[] paths = path.split("[/]+");
        if (paths.length > 1) {
            String tp = "";
            for (int i = 0; i < paths.length - 1; i++) {
                if (paths[i] != null && paths[i].trim().length() > 0) {
                    tp = tp + '/' + paths[i];
                    if (zooKeeper.exists(tp, false) == null)
                        zooKeeper.create(tp, "dir".getBytes(),
                                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                CreateMode.PERSISTENT);
                }
            }
        }
        String nodePath = zooKeeper.create(path, (ip + ":" + port).getBytes(),
                ZooDefs.Ids.READ_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);
        log.debug("zookeeper node created on '{}'", nodePath);
        zooKeeper.getData(nodePath, this, null);
        registerNode.add(nodePath);
    }
    
	public void registerNoseq(String path, String ip, String port)
			throws TException, InterruptedException, KeeperException {
		String[] paths = path.split("[/]+");
		if (paths.length > 1) {
			String tp = "";
			for (int i = 0; i < paths.length - 1; i++) {
				if (paths[i] != null && paths[i].trim().length() > 0) {
					tp = tp + '/' + paths[i];
					if (zooKeeper.exists(tp, false) == null)
						zooKeeper.create(tp, "dir".getBytes(),
								ZooDefs.Ids.OPEN_ACL_UNSAFE,
								CreateMode.PERSISTENT);
				}
			}
		}
		String nodePath = zooKeeper.create(path, (ip + ":" + port).getBytes(),
				ZooDefs.Ids.READ_ACL_UNSAFE, CreateMode.EPHEMERAL);
		log.debug("zookeeper node created on '{}'", nodePath);
		zooKeeper.getData(nodePath, this, null);
		registerNode.add(nodePath);
	}
    
    /**
     * zookeeper服务器端事件处理
     * SyncConnected状态代表与zookeeper服务器端连接成功
     * 服务器端节点变化通知也是走此状态（因为受到了通知必然连接成功）
     * 
     * Disconnected状态代表服务器端无法响应本地客户端，连接断开
     * 
     * Expired本地客户端与服务器端的连接空闲时间太长，过了timeout了
     * 
     * 
     * 但凡当前客户端watch的节点被改变了，我们就让客户端会重新连接，以防止配置文件被别人修改。
     * 
     */
	@Override
    public void process(WatchedEvent watchedEvent) {
        log.info("get Event: {}", watchedEvent);
        switch (watchedEvent.getState()) {
            case SyncConnected:
            	//If the current count equals zero then nothing happens.
                connectedSignal.countDown();
                //先处理register这种特殊事件
                if (run && watchedEvent.getPath() != null && registerNode.contains(watchedEvent.getPath())) {
                    switch (watchedEvent.getType()) {
                        case NodeDeleted:
                        case NodeDataChanged:
                            log.error("My node was modified!!! close and reconnect");
                            try {
                                reconnect();
                            } catch (Exception e) {
                                log.error("Exception when recover", e);
                                System.exit(1);
                            }
                    }
                }
                else{//普通observer的节点事件
	                switch (watchedEvent.getType()) {
	                	case NodeDeleted:
	                	case NodeDataChanged:
	                	//如果节点发生变化，我有观察，那么需要进行处理
	                	if(run && watchedEvent.getPath() != null ){
	                		try{
	                			//处理节点
	                			dealNodeChange( watchedEvent.getPath()  );
	                		}catch(Exception e){
		                		log.error("Exception when update live server memory ", e);
		                        //System.exit(1);
		                	}
	                	}
	               		break;
	                	case NodeChildrenChanged:
	                	//如果子节点发生变化，我有观察，那么需要进行处理
	            	    if(run && watchedEvent.getPath() != null){
	                        try {
	                        	dealChildrenNodeChange( watchedEvent.getPath() );
	                        } catch (Exception e) {
	                            log.error("Exception when update live server memory ", e);
	                            //System.exit(1);
	                        }
	                	}
	            	    break;
	                }
                }
        		break;
            case Disconnected:
            case Expired:
                if (run)
                    try {
                        reconnect();
                    } catch (Exception e) {
                        log.error("re connect to {}", connectStr);
                        log.error("re connect exception: ", e);
                    }
                break;
        }
    }

	/**
	 * @param callback
	 * @uml.property  name="callback"
	 */
	@SuppressWarnings("rawtypes")
	public void setCallback(Callable callback) {
        this.callback = callback;
    }
	
	public List<ZKnode> getObservServer(String zkpath){
		List<ZKnode> list = new ArrayList<ZKnode>();
		List<ZKnode> l = observeServerNodeChildrenNameDataListMap.get(zkpath);
		if( l != null){
			for(ZKnode x : l)
			list.add( x );
		}
		return list;
	}
	public HashMap<String, List<ZKnode>> getLivedObserveServer(){
		HashMap<String, List<ZKnode>> rs = new HashMap<String, List<ZKnode>>();
		
		for(String x : observeServerNonLeafNode){
			rs.put(x, getObservServer(x));
		}
		return rs;
	}

    /**
	 * @uml.property  name="connectStr"
	 */
    private String connectStr;
    /**
	 * @uml.property  name="sessionTimeOut"
	 */
    private int sessionTimeOut;
    /**
	 * @uml.property  name="run"
	 */
    private boolean run = true;
    /**
	 * @uml.property  name="zooKeeper"
	 * @uml.associationEnd  
	 */
    private ZooKeeper zooKeeper;
    /**
	 * @uml.property  name="connectedSignal"
	 */
    private CountDownLatch connectedSignal;
    
	/**
	 * @uml.property  name="callback"
	 */
	@SuppressWarnings("rawtypes")
	public Callable callback;
    /**
	 * @uml.property  name="registerNode"
	 * @uml.associationEnd  multiplicity="(0 -1)" elementType="java.lang.String"
	 */
    private HashSet<String> registerNode = new HashSet<String>();
    private static Logger log = LoggerFactory.getLogger(ZK.class);
    
    //记录要监听的zookeeper中某path，
    /**
	 * @uml.property  name="observeServerNonLeafNode"
	 * @uml.associationEnd  multiplicity="(0 -1)" elementType="java.lang.String"
	 */
    private HashSet<String> observeServerNonLeafNode = new HashSet<String>();    
    /**
	 * @uml.property  name="observeServerNonLeafNodeDealMap"
	 * @uml.associationEnd  qualifier="path:java.lang.String com.zhongsou.weibo.zk.ZKObserveDealer"
	 */
    private HashMap<String, ZKObserveDealer> observeServerNonLeafNodeDealMap = new HashMap<String, ZKObserveDealer>();
    //path以及path下的子节点数据列表
    /**
	 * @uml.property  name="observeServerNodeChildrenNameDataListMap"
	 * @uml.associationEnd  qualifier="zkpath:java.lang.String java.util.List"
	 */
    private HashMap<String, List<ZKnode>> observeServerNodeChildrenNameDataListMap = new HashMap<String, List<ZKnode>>();
    
    /**
	 * @uml.property  name="observeServerLeafNode"
	 * @uml.associationEnd  multiplicity="(0 -1)" elementType="java.lang.String"
	 */
    private HashSet<String> observeServerLeafNode = new HashSet<String>();
    /**
	 * @uml.property  name="observeServerLeafNodeDealMap"
	 * @uml.associationEnd  qualifier="path:java.lang.String com.zhongsou.weibo.zk.ZKObserveDealer"
	 */
    private HashMap<String, ZKObserveDealer> observeServerLeafNodeDealMap = new HashMap<String, ZKObserveDealer> ();
    /**
	 * @uml.property  name="observeServerNodeNameDataMap"
	 * @uml.associationEnd  qualifier="path:java.lang.String com.zhongsou.weibo.zk.ZKnode"
	 */
    private HashMap<String, ZKnode> observeServerNodeNameDataMap = new HashMap<String, ZKnode>();
}
